package com.lvmama.rhino.analyze.client

import com.lvmama.rhino.common.entity.{AIMachineReceipt, JobConfig}
import com.lvmama.rhino.common.utils.JDBCUtil.JDBCTemplate
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions._
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.StringType
import org.apache.spark.storage.StorageLevel

/**
  * Created by yuanxiaofeng on 2016/8/2.
  */
class AIMachineStat(config: JobConfig, path: String, k: Int) extends SparkApplication{
  override var appName: String = "AIMachineStatJob"
  override var sparkConfig: Map[String, String] = config.spark

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)
    sparkConfig += ("spark.kryo.registrator" -> "com.lvmama.rhino.common.utils.serialization.AIStatRegistrator")
    sparkConfig += ("spark.master" -> "local[3]")
    withSparkContext{ sc =>

      val JDBCDefault = JDBCTemplate.JDBCDefaultSet
      val connP = JDBCTemplate.getConProperties

      val hiveCtx = new HiveContext(sc)
      val bc = sc.broadcast[Int](k)
      import hiveCtx.implicits._
      val aistat = sc.textFile(path).map(l => AIMachineReceipt(AIMachineReceipt.unapply(l))).toDF()
      val forwardEvent = aistat.repartition(3)
        .filter(aistat("lt").equalTo("forward"))
        .select($"ad", $"pc")
        .groupBy($"ad", $"pc")
        .agg(count("*") as "counts")
        .withColumn("page_codes", concat_ws(":", $"pc", $"counts"))
        .groupBy($"ad")
        .agg(collect_list($"page_codes").cast(StringType).alias("page_codes"))

      val clickEvent = aistat.filter(aistat("lt").equalTo("click")).repartition(3)
      clickEvent.persist(StorageLevel.MEMORY_AND_DISK_SER)
      val buttonClick = clickEvent.select($"ad", $"ei")
        .groupBy($"ad", $"ei")
        .agg(count("*") as "counts")
        .withColumn("button_clicks", concat_ws(":", col("ei"), col("counts")))
        .groupBy($"ad")
        .agg(collect_list($"button_clicks").cast(StringType).alias("button_clicks"))

      val windowSpec = Window.partitionBy($"ad").orderBy($"counts".desc)

      val questionTopK = clickEvent.filter(col("ei").equalTo("nq"))
        .select($"ad", $"qn")
        .groupBy($"ad", $"qn")
        .agg(count("*") as "counts")
        .select($"ad",$"qn",$"counts", rank().over(windowSpec).as("row_number")).where($"row_number" <= bc.value)
        .groupBy($"ad")
        .agg(collect_list(concat_ws(":", $"qn", $"counts")).cast(StringType).as("questions"))

      val questionTopKOfCountry = clickEvent.filter(col("ei").equalTo("nq"))
        .select(col("qn"))
        .groupBy(col("qn"))
        .agg(count("*") as "counts")
        .sort(col("counts").desc)
        .limit(bc.value)
        .withColumn("questions", concat_ws(":", col("qn"), col("counts")))
        .withColumn("ad", lit("全国"))
        .groupBy($"ad")
        .agg(collect_list($"questions").cast(StringType).alias("questions"))

      val sourceVisit = clickEvent.filter(col("ei").equalTo("pp"))
        .select(col("ad"), col("pc"))
        .groupBy(col("ad"), col("pc"))
        .agg(count("*") as "counts")
        .withColumn("page_codes", concat_ws(":", col("pc"), col("counts")))
        .groupBy($"ad")
        .agg(collect_list($"page_codes").cast(StringType).alias("page_codes"))

      val aiAskAnswer = clickEvent.filter(col("ei").equalTo("aq"))
        .select(col("ad"))
        .groupBy(col("ad"))
        .agg(count("*") as "counts")

      forwardEvent.select(col("page_codes"), col("ad") as "province").repartition(1).write.mode(SaveMode.Append)
        .option("driver", JDBCDefault.get("driver").get).jdbc(JDBCDefault.get("conn").get, "ai_page_visit" , connP)
      buttonClick.select(col("ad") as "province", col("button_clicks")).repartition(1).write.mode(SaveMode.Append)
        .option("driver", JDBCDefault.get("driver").get).jdbc(JDBCDefault.get("conn").get, "ai_button_click" , connP)
      questionTopK.select(col("ad") as "province", col("questions")).repartition(1).write.mode(SaveMode.Append)
        .option("driver", JDBCDefault.get("driver").get).jdbc(JDBCDefault.get("conn").get, "ai_frequently_question" , connP)
      questionTopKOfCountry.select(col("ad") as "province", col("questions")).repartition(1).write.mode(SaveMode.Append)
        .option("driver", JDBCDefault.get("driver").get).jdbc(JDBCDefault.get("conn").get, "ai_frequently_question" , connP)
      sourceVisit.select(col("ad") as "province", col("page_codes")).repartition(1).write.mode(SaveMode.Append)
        .option("driver", JDBCDefault.get("driver").get).jdbc(JDBCDefault.get("conn").get, "ai_source_visit" , connP)
      aiAskAnswer.select(col("ad") as "province", col("counts")).repartition(1).write.mode(SaveMode.Append)
        .option("driver", JDBCDefault.get("driver").get).jdbc(JDBCDefault.get("conn").get, "ai_ask_answer_stat" , connP)

    }
  }
}

object AIMachineStat{
  def apply(path: String, k: Int): Unit = {
    val config = JobConfig()
    new AIMachineStat(config, path, k).execute()
  }
}
